package com.dnjn.happiness.score.amqp;

import com.dnjn.happiness.common.amqp.AmqpConst;

import com.dnjn.happiness.common.amqp.AmqpSender;
import com.dnjn.happiness.common.core.exception.base.BaseException;
import com.dnjn.happiness.common.core.utils.JsonUtils;
import com.dnjn.happiness.common.core.score.dto.HappyDTO;
import com.dnjn.happiness.common.core.score.vo.HappyVO;
import com.dnjn.happiness.score.service.IHScoresService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.util.StopWatch;

import java.io.IOException;

@Slf4j
@Component
public class HappyAmqpReveiver {
    @Autowired
    RabbitTemplate rabbitTemplate;

    @Autowired
    private IHScoresService happinessScoresService;


    @RabbitListener(queues = AmqpConst.SCORE_QUEUE_NAME)
    public void receiveAndConvert(/*@Payload*/ HappyVO happyVO, Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            StopWatch stopWatch = new StopWatch("happyscore");
            stopWatch.start();
            HappyDTO happyDTO = handleAmqpMessage(happyVO, message,channel,deliveryTag);
            // 发送回队列
            //rabbitTemplate.convertAndSend(AmqpConst.HAPPY_QUEUE_EXCHANGE_NAME_SEND, AmqpConst.HAPPY_QUEUE_ROUTING_KEY_SEND, happyDTO);
            AmqpSender.sendUserScore(happyDTO);
            stopWatch.stop();
            log.info("happy time spent {}s",stopWatch.getTotalTimeSeconds());
            channel.basicAck(deliveryTag, false);
        } catch (BaseException e) {
            log.error(e.getMessage(), e);
            throw e;
        } catch (Exception e) {
            log.error(e.getMessage(), e);

            try {
                // basicReject一次只能拒绝接收一个消息，可以设置是否requeue。
                // basicNack方法可以支持一次0个或多个消息的拒收，可以设置是否requeue
                channel.basicNack(deliveryTag, false, true);
                // log.info("consumer receive this message: {},reject it", JsonKit.toJson(message.getBody()));
                // channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
                //channel.basicReject(deliveryTag, true);

            } catch (IOException ee) {
                log.error("exception:{}", ee.getMessage());
            }
        }
    }
//    public void receiveAndConvert(/*@Payload*/ HappyVO ferSeed, Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException {
//        channel.basicAck(deliveryTag,false);
//    }


    private HappyDTO handleAmqpMessage(HappyVO vo, Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws Exception {
        log.info("amqp_consumer.receive.message: {} ", JsonUtils.toJsonString(vo));

        if (null ==vo.getUserId()) {
            channel.basicAck( deliveryTag,false);
            throw new BaseException("invalid.userId");
        }

        HappyDTO happinessScores = happinessScoresService.getHappinessScores(vo);
        return happinessScores;


    }
}
